//! 事件分发器
//! 
//! 提供高性能的事件分发和处理机制，支持多种分发策略

// use crate::event::Event;  // 已注释：未使用的导入
// use crate::event::EventContext;  // 已注释：未使用的导入
use crate::event::{
    EventPayload, DynamicEvent, DynamicEventHandler, EventResult, 
    EventBatch, Priority, EventQueue, SafeAtomicBool, SafeAtomicU64
};
// use crate::core::concurrency::{LockFreeMPMCQueue, BoundedMPMCQueue};  // 已注释：未使用的导入
use crate::core::concurrency::LockFreeMPMCQueue;
use crate::core::concurrency::queue::MPMCQueue;
// use crate::core::concurrency::BoundedMPMCQueue;

use std::sync::{Arc, RwLock, Mutex};
use fxhash::FxHashMap;
use std::thread;
use std::time::{Duration, Instant};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

// =============================================================================
// 局部类型别名（为该模块特定的复杂类型）
// =============================================================================

/// 包装器处理器映射
type WrapperHandlerMap = Arc<RwLock<FxHashMap<String, Vec<Arc<HandlerWrapper>>>>>;

/// 优先级队列类型
type PriorityEventQueue = Arc<Mutex<std::collections::BinaryHeap<PriorityQueueItem>>>;

/// 通配符处理器列表
type WildcardHandlerList = Arc<RwLock<Vec<Arc<HandlerWrapper>>>>;

// =============================================================================
// 常量和枚举定义
// =============================================================================

/// 事件分发策略
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum DispatchStrategy {
    /// 轮询分发 - 依次分发给处理器
    RoundRobin,
    /// 广播分发 - 发送给所有匹配的处理器
    Broadcast,
    /// 优先级分发 - 根据处理器优先级选择
    Priority,
    /// 负载均衡分发 - 选择负载最轻的处理器
    LoadBalance,
}

/// 分发器配置
#[derive(Debug, Clone)]
pub struct DispatcherConfig {
    /// 工作线程数
    pub worker_count: usize,
    /// 事件队列容量
    pub queue_capacity: usize,
    /// 批处理大小
    pub batch_size: usize,
    /// 批处理超时（毫秒）
    pub batch_timeout_ms: u64,
    /// 分发策略
    pub strategy: DispatchStrategy,
    /// 是否启用优先级队列
    pub priority_queue: bool,
    /// 最大重试次数
    pub max_retry_count: u32,
    /// 重试延迟（毫秒）
    pub retry_delay_ms: u64,
    /// 最大重试尝试次数（新增）
    pub max_retry_attempts: u32,
}

impl Default for DispatcherConfig {
    fn default() -> Self {
        Self {
            worker_count: num_cpus::get(),
            queue_capacity: 10000,
            batch_size: 100,
            batch_timeout_ms: 100,
            strategy: DispatchStrategy::RoundRobin,
            priority_queue: true,
            max_retry_count: 3,
            retry_delay_ms: 1000,
            max_retry_attempts: 3,
        }
    }
}

/// 处理器包装器，包含处理器和其统计信息
struct HandlerWrapper {
    handler: Arc<dyn DynamicEventHandler>,
    /// 处理的事件数
    processed_count: AtomicU64,
    /// 失败的事件数  
    failed_count: AtomicU64,
    /// 总处理时间（毫秒）
    total_processing_time: AtomicU64,
    /// 当前负载（正在处理的事件数）
    current_load: AtomicU64,
}

impl HandlerWrapper {
    fn new(handler: Arc<dyn DynamicEventHandler>) -> Self {
        Self {
            handler,
            processed_count: AtomicU64::new(0),
            failed_count: AtomicU64::new(0),
            total_processing_time: AtomicU64::new(0),
            current_load: AtomicU64::new(0),
        }
    }

    fn handle_event(&self, event: &DynamicEvent) -> EventResult {
        let start_time = Instant::now();
        self.current_load.fetch_add(1, Ordering::Relaxed);
        
        let result = self.handler.handle_dynamic(event);
        
        let processing_time = start_time.elapsed().as_millis() as u64;
        self.total_processing_time.fetch_add(processing_time, Ordering::Relaxed);
        self.current_load.fetch_sub(1, Ordering::Relaxed);
        
        match result {
            EventResult::Success => {
                self.processed_count.fetch_add(1, Ordering::Relaxed);
            }
            EventResult::Failed(_) => {
                self.failed_count.fetch_add(1, Ordering::Relaxed);
            }
            _ => {}
        }
        
        result
    }

    fn get_load(&self) -> u64 {
        self.current_load.load(Ordering::Relaxed)
    }

    fn get_stats(&self) -> HandlerStats {
        HandlerStats {
            name: self.handler.name().to_string(),
            processed_count: self.processed_count.load(Ordering::Relaxed),
            failed_count: self.failed_count.load(Ordering::Relaxed),
            total_processing_time: self.total_processing_time.load(Ordering::Relaxed),
            current_load: self.current_load.load(Ordering::Relaxed),
        }
    }
}

/// 处理器统计信息
#[derive(Debug, Clone)]
pub struct HandlerStats {
    pub name: String,
    pub processed_count: u64,
    pub failed_count: u64,
    pub total_processing_time: u64,
    pub current_load: u64,
}

impl HandlerStats {
    /// 获取成功率
    pub fn success_rate(&self) -> f64 {
        if self.processed_count + self.failed_count == 0 {
            1.0
        } else {
            self.processed_count as f64 / (self.processed_count + self.failed_count) as f64
        }
    }

    /// 获取平均处理时间
    pub fn avg_processing_time(&self) -> f64 {
        if self.processed_count == 0 {
            0.0
        } else {
            self.total_processing_time as f64 / self.processed_count as f64
        }
    }
}

/// 优先级队列项
#[derive(Debug)]
struct PriorityQueueItem {
    event: DynamicEvent,
    priority: Priority,
    enqueue_time: Instant,
}

/// Worker线程上下文（封装参数以减少复杂性）
#[derive(Clone)]
struct WorkerContext {
    worker_id: usize,
    running: SafeAtomicBool,
    event_queue: EventQueue<DynamicEvent>,
    priority_queue: PriorityEventQueue,
    handlers: WrapperHandlerMap,
    wildcard_handlers: WildcardHandlerList,
    config: DispatcherConfig,
    round_robin_counter: SafeAtomicU64,
}

impl WorkerContext {
    fn new(
        worker_id: usize,
        running: SafeAtomicBool,
        event_queue: EventQueue<DynamicEvent>,
        priority_queue: PriorityEventQueue,
        handlers: WrapperHandlerMap,
        wildcard_handlers: WildcardHandlerList,
        config: DispatcherConfig,
        round_robin_counter: SafeAtomicU64,
    ) -> Self {
        Self {
            worker_id,
            running,
            event_queue,
            priority_queue,
            handlers,
            wildcard_handlers,
            config,
            round_robin_counter,
        }
    }
}

impl PartialEq for PriorityQueueItem {
    fn eq(&self, other: &Self) -> bool {
        self.priority == other.priority
    }
}

impl Eq for PriorityQueueItem {}

impl PartialOrd for PriorityQueueItem {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for PriorityQueueItem {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        // 注意：BinaryHeap是最大堆，这里要反转比较以实现优先级队列
        other.priority.cmp(&self.priority)
            .then_with(|| self.enqueue_time.cmp(&other.enqueue_time))
    }
}

/// 事件发送器（线程安全的发布接口）
#[derive(Clone)]
pub struct EventSender {
    /// 事件队列
    event_queue: EventQueue<DynamicEvent>,
    /// 优先级队列
    priority_queue: PriorityEventQueue,
    /// 配置信息
    config: DispatcherConfig,
    /// 统计信息
    stats: Arc<DispatcherStats>,
}

impl EventSender {
    /// 发布事件
    pub fn publish(&self, event: DynamicEvent) -> Result<(), DynamicEvent> {
        self.stats.total_events.fetch_add(1, Ordering::Relaxed);
        
        if self.config.priority_queue && event.priority() > Priority::Normal {
            // 高优先级事件进入优先级队列
            let mut pq = self.priority_queue.lock().unwrap();
            pq.push(PriorityQueueItem {
                priority: event.priority(),
                enqueue_time: Instant::now(),
                event,
            });
            drop(pq);
        } else {
            // 普通事件进入常规队列
            self.event_queue.enqueue(event)?;
        }
        
        self.stats.queued_events.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    /// 批量发布事件
    pub fn publish_batch(&self, events: Vec<DynamicEvent>) -> Result<(), Vec<DynamicEvent>> {
        let mut failed_events = Vec::new();
        
        for event in events {
            if let Err(failed_event) = self.publish(event) {
                failed_events.push(failed_event);
            }
        }
        
        if failed_events.is_empty() {
            Ok(())
        } else {
            Err(failed_events)
        }
    }
}

/// 事件分发器
pub struct EventDispatcher {
    /// 配置
    config: DispatcherConfig,
    /// 事件队列
    event_queue: EventQueue<DynamicEvent>,
    /// 优先级队列
    priority_queue: PriorityEventQueue,
    /// 处理器映射 (事件类型 -> 处理器列表)
    handlers: WrapperHandlerMap,
    /// 通配符处理器（处理所有事件类型）
    wildcard_handlers: WildcardHandlerList,
    /// 运行状态
    running: SafeAtomicBool,
    /// 工作线程句柄
    worker_handles: Vec<thread::JoinHandle<()>>,
    /// 轮询计数器（用于轮询分发）
    round_robin_counter: SafeAtomicU64,
    /// 分发器统计
    stats: Arc<DispatcherStats>,
}

impl EventDispatcher {
    /// 创建新的事件分发器
    pub fn new(config: DispatcherConfig) -> Self {
        let running = Arc::new(AtomicBool::new(false));
        let event_queue = Arc::new(LockFreeMPMCQueue::new());
        let priority_queue = Arc::new(Mutex::new(std::collections::BinaryHeap::new()));
        let handlers = Arc::new(RwLock::new(FxHashMap::default()));
        let wildcard_handlers = Arc::new(RwLock::new(Vec::new()));
        let round_robin_counter = Arc::new(AtomicU64::new(0));
        let stats = Arc::new(DispatcherStats::new());

        Self {
            config,
            event_queue,
            priority_queue,
            handlers,
            wildcard_handlers,
            running,
            worker_handles: Vec::new(),
            round_robin_counter,
            stats,
        }
    }

    /// 注册事件处理器
    pub fn register_handler(&mut self, event_type: &str, handler: Arc<dyn DynamicEventHandler>) {
        let wrapper = Arc::new(HandlerWrapper::new(handler));
        let mut handlers = self.handlers.write().unwrap();
        
        if event_type == "*" {
            // 通配符处理器
            let mut wildcard_handlers = self.wildcard_handlers.write().unwrap();
            wildcard_handlers.push(wrapper);
        } else {
            // 特定类型处理器
            handlers.entry(event_type.to_string())
                .or_default()
                .push(wrapper);
        }
    }

    /// 取消注册处理器
    pub fn unregister_handler(&mut self, event_type: &str, handler_name: &str) {
        let mut handlers = self.handlers.write().unwrap();
        
        if event_type == "*" {
            let mut wildcard_handlers = self.wildcard_handlers.write().unwrap();
            wildcard_handlers.retain(|h| h.handler.name() != handler_name);
        } else if let Some(handler_list) = handlers.get_mut(event_type) {
            handler_list.retain(|h| h.handler.name() != handler_name);
            if handler_list.is_empty() {
                handlers.remove(event_type);
            }
        }
    }

    /// 创建线程安全的发送器
    pub fn clone_sender(&self) -> EventSender {
        EventSender {
            event_queue: self.event_queue.clone(),
            priority_queue: self.priority_queue.clone(),
            config: self.config.clone(),
            stats: self.stats.clone(),
        }
    }

    /// 启动分发器
    pub fn start(&mut self) {
        if self.running.load(Ordering::Relaxed) {
            return;
        }

        self.running.store(true, Ordering::Relaxed);
        
        // 启动工作线程
        for worker_id in 0..self.config.worker_count {
            let running = self.running.clone();
            let event_queue = self.event_queue.clone();
            let priority_queue = self.priority_queue.clone();
            let handlers = self.handlers.clone();
            let wildcard_handlers = self.wildcard_handlers.clone();
            let config = self.config.clone();
            let round_robin_counter = self.round_robin_counter.clone();
            let _stats = DispatcherStats::new(); // 每个worker有自己的stats引用

            let handle = thread::Builder::new()
                .name(format!("event-worker-{worker_id}"))
                .spawn(move || {
                    let context = WorkerContext::new(
                        worker_id,
                        running,
                        event_queue,
                        priority_queue,
                        handlers,
                        wildcard_handlers,
                        config,
                        round_robin_counter,
                    );
                    Self::worker_loop(context);
                })
                .expect("Failed to spawn worker thread");

            self.worker_handles.push(handle);
        }
    }

    /// 停止分发器
    pub fn stop(&mut self) {
        self.running.store(false, Ordering::Relaxed);
        
        // 等待所有工作线程结束
        while let Some(handle) = self.worker_handles.pop() {
            let _ = handle.join();
        }
    }

    /// 发布事件（委托给内部发送器）
    pub fn publish(&self, event: DynamicEvent) -> Result<(), DynamicEvent> {
        let sender = self.clone_sender();
        sender.publish(event)
    }

    /// 批量发布事件（委托给内部发送器）
    pub fn publish_batch(&self, events: Vec<DynamicEvent>) -> Result<(), Vec<DynamicEvent>> {
        let sender = self.clone_sender();
        sender.publish_batch(events)
    }

    /// 获取分发器统计信息
    pub fn get_stats(&self) -> DispatcherStatsSnapshot {
        DispatcherStatsSnapshot {
            total_events: self.stats.total_events.load(Ordering::Relaxed),
            processed_events: self.stats.processed_events.load(Ordering::Relaxed),
            failed_events: self.stats.failed_events.load(Ordering::Relaxed),
            queued_events: self.stats.queued_events.load(Ordering::Relaxed),
            processing_rate: self.stats.processing_rate(),
            success_rate: self.stats.success_rate(),
            uptime_seconds: self.stats.start_time.elapsed().as_secs(),
        }
    }

    /// 获取处理器统计信息
    pub fn get_handler_stats(&self) -> Vec<HandlerStats> {
        let mut stats = Vec::new();
        
        // 特定类型处理器
        let handlers = self.handlers.read().unwrap();
        for handler_list in handlers.values() {
            for wrapper in handler_list {
                stats.push(wrapper.get_stats());
            }
        }
        
        // 通配符处理器
        let wildcard_handlers = self.wildcard_handlers.read().unwrap();
        for wrapper in wildcard_handlers.iter() {
            stats.push(wrapper.get_stats());
        }
        
        stats
    }

    /// 工作线程主循环（优化参数）
    fn worker_loop(context: WorkerContext) {
        let mut batch = EventBatch::new(
            format!("worker-batch-{}", context.worker_id), 
            context.config.batch_size
        );
        let mut last_batch_time = Instant::now();

        while context.running.load(Ordering::Relaxed) {
            let mut processed_any = false;

            // 处理优先级队列中的事件
            if context.config.priority_queue {
                if let Ok(mut pq) = context.priority_queue.try_lock() {
                    if let Some(priority_item) = pq.pop() {
                        drop(pq);
                        Self::process_single_event(
                            priority_item.event,
                            &context.handlers,
                            &context.wildcard_handlers,
                            &context.config,
                            &context.round_robin_counter,
                        );
                        processed_any = true;
                    }
                }
            }

            // 处理常规队列中的事件
            if let Some(event) = context.event_queue.dequeue() {
                if context.config.batch_size > 1 {
                    // 批处理模式
                    if batch.add_event(event).is_err() {
                        // 批次已满，处理批次
                        Self::process_event_batch(
                            &mut batch,
                            &context.handlers,
                            &context.wildcard_handlers,
                            &context.config,
                            &context.round_robin_counter,
                        );
                        last_batch_time = Instant::now();
                    }
                } else {
                    // 单事件处理模式
                    Self::process_single_event(
                        event,
                        &context.handlers,
                        &context.wildcard_handlers,
                        &context.config,
                        &context.round_robin_counter,
                    );
                }
                processed_any = true;
            }

            // 检查批次超时
            if !batch.is_empty() 
                && last_batch_time.elapsed().as_millis() >= context.config.batch_timeout_ms as u128 
            {
                Self::process_event_batch(
                    &mut batch,
                    &context.handlers,
                    &context.wildcard_handlers,
                    &context.config,
                    &context.round_robin_counter,
                );
                last_batch_time = Instant::now();
            }

            // 如果没有处理任何事件，短暂休眠避免CPU占用过高
            if !processed_any {
                thread::sleep(Duration::from_millis(1));
            }
        }

        // 处理剩余的批次事件
        if !batch.is_empty() {
            Self::process_event_batch(
                &mut batch,
                &context.handlers,
                &context.wildcard_handlers,
                &context.config,
                &context.round_robin_counter,
            );
        }
    }

    /// 处理单个事件
    fn process_single_event(
        event: DynamicEvent,
        handlers: &Arc<RwLock<FxHashMap<String, Vec<Arc<HandlerWrapper>>>>>,
        wildcard_handlers: &Arc<RwLock<Vec<Arc<HandlerWrapper>>>>,
        config: &DispatcherConfig,
        round_robin_counter: &Arc<AtomicU64>,
    ) {
        let event_type = event.event_type();
        let handlers_guard = handlers.read().unwrap();
        let wildcard_guard = wildcard_handlers.read().unwrap();

        // 获取匹配的处理器
        let mut target_handlers = Vec::new();

        // 添加特定类型处理器
        if let Some(type_handlers) = handlers_guard.get(event_type) {
            target_handlers.extend(type_handlers.iter().cloned());
        }

        // 添加通配符处理器
        target_handlers.extend(wildcard_guard.iter().cloned());

        drop(handlers_guard);
        drop(wildcard_guard);

        if target_handlers.is_empty() {
            return;
        }

        // 根据策略选择处理器
        let selected_handlers = Self::select_handlers(&target_handlers, config, round_robin_counter);

        // 处理事件
        for handler in selected_handlers {
            let result = handler.handle_event(&event);
            
            // 可以在这里添加重试逻辑、错误处理等
            match result {
                EventResult::Success => {
                    // 处理成功，继续下一个处理器（如果是广播模式）
                    if config.strategy != DispatchStrategy::Broadcast {
                        break; // 非广播模式，处理成功后停止
                    }
                }
                EventResult::Failed(_) => {
                    // 处理失败，可以添加重试逻辑
                }
                EventResult::Retry(_) => {
                    // 需要重试，可以重新入队或延迟处理
                }
                EventResult::Skipped => {
                    // 跳过处理，继续下一个处理器
                }
            }
        }
    }

    /// 处理事件批次
    fn process_event_batch(
        batch: &mut EventBatch<Box<dyn EventPayload>>,
        handlers: &Arc<RwLock<FxHashMap<String, Vec<Arc<HandlerWrapper>>>>>,
        wildcard_handlers: &Arc<RwLock<Vec<Arc<HandlerWrapper>>>>,
        config: &DispatcherConfig,
        round_robin_counter: &Arc<AtomicU64>,
    ) {
        for event in batch.events.drain(..) {
            Self::process_single_event(event, handlers, wildcard_handlers, config, round_robin_counter);
        }
        batch.clear();
    }

    /// 根据策略选择处理器
    fn select_handlers(
        handlers: &[Arc<HandlerWrapper>],
        config: &DispatcherConfig,
        round_robin_counter: &Arc<AtomicU64>,
    ) -> Vec<Arc<HandlerWrapper>> {
        match config.strategy {
            DispatchStrategy::Broadcast => {
                // 广播：返回所有处理器
                handlers.to_vec()
            }
            DispatchStrategy::RoundRobin => {
                // 轮询：选择一个处理器
                if handlers.is_empty() {
                    vec![]
                } else {
                    let index = round_robin_counter.fetch_add(1, Ordering::Relaxed) % handlers.len() as u64;
                    vec![handlers[index as usize].clone()]
                }
            }
            DispatchStrategy::Priority => {
                // 优先级：选择优先级最高的处理器
                let mut sorted_handlers = handlers.to_vec();
                sorted_handlers.sort_by(|a, b| {
                    b.handler.handler_priority().cmp(&a.handler.handler_priority())
                });
                if sorted_handlers.is_empty() {
                    vec![]
                } else {
                    vec![sorted_handlers[0].clone()]
                }
            }
            DispatchStrategy::LoadBalance => {
                // 负载均衡：选择负载最轻的处理器
                let min_load_handler = handlers.iter()
                    .min_by_key(|h| h.get_load())
                    .cloned();
                min_load_handler.map(|h| vec![h]).unwrap_or_default()
            }
        }
    }
}

impl Drop for EventDispatcher {
    fn drop(&mut self) {
        self.stop();
    }
}

/// 分发器统计信息
#[derive(Debug)]
pub struct DispatcherStats {
    /// 总事件数
    total_events: AtomicU64,
    /// 已处理事件数
    processed_events: AtomicU64,
    /// 失败事件数
    failed_events: AtomicU64,
    /// 队列中的事件数
    queued_events: AtomicU64,
    /// 启动时间
    start_time: Instant,
}

impl DispatcherStats {
    fn new() -> Self {
        Self {
            total_events: AtomicU64::new(0),
            processed_events: AtomicU64::new(0),
            failed_events: AtomicU64::new(0),
            queued_events: AtomicU64::new(0),
            start_time: Instant::now(),
        }
    }

    /// 获取处理速率（事件/秒）
    pub fn processing_rate(&self) -> f64 {
        let elapsed = self.start_time.elapsed().as_secs_f64();
        if elapsed > 0.0 {
            self.processed_events.load(Ordering::Relaxed) as f64 / elapsed
        } else {
            0.0
        }
    }

    /// 获取成功率
    pub fn success_rate(&self) -> f64 {
        let total = self.total_events.load(Ordering::Relaxed);
        if total == 0 {
            1.0
        } else {
            self.processed_events.load(Ordering::Relaxed) as f64 / total as f64
        }
    }
}

/// 分发器统计信息快照
#[derive(Debug, Clone)]
pub struct DispatcherStatsSnapshot {
    pub total_events: u64,
    pub processed_events: u64,
    pub failed_events: u64,
    pub queued_events: u64,
    pub processing_rate: f64,
    pub success_rate: f64,
    pub uptime_seconds: u64,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::event::{StringPayload, EventMetadata, Event};
    use std::sync::atomic::AtomicU32;

    /// 测试用的简单事件处理器
    struct TestHandler {
        name: String,
        processed_count: AtomicU32,
    }

    impl TestHandler {
        fn new(name: &str) -> Self {
            Self {
                name: name.to_string(),
                processed_count: AtomicU32::new(0),
            }
        }

        fn get_processed_count(&self) -> u32 {
            self.processed_count.load(Ordering::Relaxed)
        }
    }

    impl DynamicEventHandler for TestHandler {
        fn handle_dynamic(&self, _event: &DynamicEvent) -> EventResult {
            self.processed_count.fetch_add(1, Ordering::Relaxed);
            thread::sleep(Duration::from_millis(10)); // 模拟处理时间
            EventResult::Success
        }

        fn name(&self) -> &str {
            &self.name
        }

        fn can_handle(&self, _event_type: &str) -> bool {
            true
        }
    }

    #[test]
    fn test_dispatcher_basic() {
        let config = DispatcherConfig {
            worker_count: 2,
            batch_size: 1, // 不使用批处理
            ..Default::default()
        };

        let mut dispatcher = EventDispatcher::new(config);
        let handler = Arc::new(TestHandler::new("test_handler"));
        
        dispatcher.register_handler("test_event", handler.clone());
        dispatcher.start();

        // 发送一些事件
        for i in 0..10 {
            let event = Event::new("test_event", StringPayload(format!("message_{i}")));
            let dynamic_event: DynamicEvent = Event::with_metadata(
                (*event.metadata).clone(),
                Box::new(event.payload)
            );
            dispatcher.publish(dynamic_event).unwrap();
        }

        // 等待处理完成
        thread::sleep(Duration::from_millis(200));
        
        let stats = dispatcher.get_stats();
        assert_eq!(stats.total_events, 10);
        
        dispatcher.stop();
        
        // 验证处理器收到了事件
        assert!(handler.get_processed_count() > 0);
    }

    #[test]
    fn test_dispatcher_priority() {
        let config = DispatcherConfig {
            worker_count: 1,
            priority_queue: true,
            ..Default::default()
        };

        let mut dispatcher = EventDispatcher::new(config);
        let handler = Arc::new(TestHandler::new("priority_handler"));
        
        dispatcher.register_handler("*", handler.clone());
        dispatcher.start();

        // 发送不同优先级的事件
        let low_event = Event::with_metadata(
            EventMetadata::new("low_priority").with_priority(Priority::Low),
            Box::new(StringPayload("low".to_string())) as Box<dyn EventPayload>
        );

        let high_event = Event::with_metadata(
            EventMetadata::new("high_priority").with_priority(Priority::Critical),
            Box::new(StringPayload("high".to_string())) as Box<dyn EventPayload>
        );

        // 先发送低优先级，再发送高优先级
        dispatcher.publish(low_event).unwrap();
        dispatcher.publish(high_event).unwrap();

        // 等待处理完成
        thread::sleep(Duration::from_millis(100));
        
        dispatcher.stop();
        
        // 高优先级事件应该被优先处理
        assert_eq!(handler.get_processed_count(), 2);
    }

    #[test]
    fn test_dispatcher_stats() {
        let mut dispatcher = EventDispatcher::new(DispatcherConfig::default());
        let handler = Arc::new(TestHandler::new("stats_handler"));
        
        dispatcher.register_handler("test", handler);
        dispatcher.start();

        // 发送事件
        for i in 0..5 {
            let event = Event::new("test", StringPayload(format!("msg_{i}")));
            let dynamic_event: DynamicEvent = Event::with_metadata(
                (*event.metadata).clone(),
                Box::new(event.payload)
            );
            dispatcher.publish(dynamic_event).unwrap();
        }

        thread::sleep(Duration::from_millis(100));
        
        let stats = dispatcher.get_stats();
        assert_eq!(stats.total_events, 5);
        
        let handler_stats = dispatcher.get_handler_stats();
        assert!(!handler_stats.is_empty());
        
        dispatcher.stop();
    }
}